4주차. 검색 고도화 — 부모-자식 분할부터 로컬 리랭킹까지 프로덕션 검색 파이프라인

출처: 실습 노트북 practice4-1~practice4-7 (신뢰 1순위, 실제 실행 코드) + 원서 『RAG 마스터: 랭체인으로 완성하는 LLM 서비스』(프리렉 2025) 4장 (개념 backbone)

기본 RAG는 질문을 그대로 임베딩해 단일 유사도로 문서를 꺼내 온다. 이 주차는 그 파이프라인의 각 단계 — 분할·질의 변형·검색 알고리즘·후처리 — 를 어떻게 고도화하는지 다룬다. 원서는 개념을 코랩에서 인메모리(Chroma·FAISS·BM25Retriever)·상용 API(gpt-4o)로 보여 주지만, 이 강의의 노트북은 같은 개념을 프로덕션 인프라(PostgreSQL/pgvector·로컬 LLM·로컬 리랭커)로 다시 구현한다. "개념은 책에서, 운영 가능한 실체는 노트북에서"가 이 주차의 핵심 가치다.

학습 목표

이 주차를 끝내면 다음을 할 수 있다.

부모-자식 분할의 동작 원리를 설명하고, 부모 문서를 RAM(InMemoryStore) 대신 영속 저장소(Chroma)에 담는 BaseStore 구현체를 만든다.
MultiQueryRetriever와 HyDE의 접근 차이를 비교하고, 로컬 추론 모델(<think> 토큰)에서 이 둘을 동작시킬 때의 함정을 안다.
BM25를 인메모리 라이브러리가 아니라 PostgreSQL 트리거 + 순색인/역색인/통계 테이블로 직접 구현하고, 그 확장성 이점을 설명한다.
pgvector HNSW 인덱스의 차원 상한 문제를 MRL(Matryoshka) 임베딩의 truncation으로 푼다.
앙상블 검색(RRF)의 재현율 한계를 이해하고, 로컬 크로스 인코더/리랭커(Qwen3-Reranker)로 2단계 검색을 완성한다.

전체 흐름도

이 주차의 7개 노트북은 기본 RAG 파이프라인의 각 단계를 순서대로 고도화한다.

[ 원본 문서: How_to_invest_money.txt / 투자설명서.pdf ]
   │
   ▼
[ 1. 청킹 전략 ]  ── practice4-1
   부모-자식 분할 (자식으로 정밀 검색, 부모를 넓은 맥락으로 반환)
   책: 부모 = InMemoryStore(RAM) / 강의: 부모 = Chroma 영속
   │
   ▼
[ 2. 질의 변형 ]  ── practice4-2, practice4-3
   MultiQuery (한 질문 → N개 관점) / HyDE (가상 답변으로 검색)
   책: gpt-4o / 강의: LM Studio 로컬 추론 모델 (<think> 처리 필요)
   │
   ▼
[ 3. 검색 알고리즘 ]  ── practice4-4, practice4-5, practice4-6
   희소(BM25) + 밀집(HNSW) + 앙상블(RRF)
   책: BM25Retriever(RAM) · FAISS(파일) / 강의: pgvector 서버 영속
   │
   ▼
[ 4. 후처리 — 리랭킹 ]  ── practice4-7
   1차 후보를 넓게(k=4) 모아 2차 정밀 재순위 → top_n=2
   책: gpt-4o LLM 리랭킹 / 강의: 로컬 Qwen3-Reranker-0.6B
   │
   ▼
[ 최종 답변: RetrievalQA(chain_type="stuff") ]

세 인프라(LM Studio 로컬 LLM/임베딩, Chroma 서버, PostgreSQL/pgvector)는 전 노트북에서 공유된다. 검색 알고리즘 3종(BM25·밀집·앙상블)과 리랭킹은 모두 투자설명서.pdf(분할 후 522건)를 대상으로 같은 질문 "이 회사가 발행한 주식의 총 발행량이 어느 정도야?"를 던져 서로를 비교한다.

0. 사전 필수 용어

부모-자식 분할(Parent-Child Chunking) — 원문을 큰 부모 청크(예: 1000자)와 작은 자식 청크(예: 200자)로 이중 분할. 자식으로 정밀 검색하고, 매칭된 자식이 속한 부모를 컨텍스트로 반환한다.
MultiQuery — 하나의 질문을 LLM이 여러 관점의 하위 질의로 변형해 각각 독립 검색한 뒤 중복을 제거해 합치는 질의 변형 기법.
HyDE(Hypothetical Document Embeddings) — 질문에 대한 가상 답변 문서를 LLM으로 먼저 만들고, 그 가상 문서를 임베딩해 검색하는 기법. "질문보다 답변이 실제 문서와 더 닮았다"는 가정.
희소 검색(Sparse Retrieval) / BM25 — 단어 빈도(TF)·희소성(IDF)·문서 길이 정규화를 결합한 키워드 랭킹. 정확한 용어 일치에 강하고 의미 포착엔 약하다.
밀집 검색(Dense Retrieval) — 임베딩으로 텍스트를 고차원 벡터로 바꿔 의미 유사도로 찾는 방식. 표현이 달라도 의미가 가까우면 찾는다.
HNSW(Hierarchical Navigable Small World) — 벡터의 근사 최근접 이웃(ANN)을 빠르게 찾는 그래프 인덱스. pgvector가 지원.
MRL(Matryoshka Representation Learning) — 임베딩의 앞쪽 차원만 잘라내도 그 자체로 유효한 저차원 임베딩이 되도록 학습하는 방식. 마트료시카 인형처럼 앞부분이 전체를 요약한다.
앙상블 검색 / RRF(Reciprocal Rank Fusion) — 희소·밀집 검색 결과를 각 문서의 순위 역수 합으로 융합하는 하이브리드. 점수 체계가 달라도 순위만으로 안정적으로 합친다.
리랭킹(Reranking) — 1차 검색으로 넓게 모은 후보를 더 정교한 모델로 재평가·재정렬하는 후처리 단계.
크로스 인코더(Cross-Encoder) — 질문과 문서를 함께 인코더에 넣어 관련성 점수를 직접 산출하는 모델. 바이 인코더(각각 인코딩)보다 정확하지만 느리다.
LM Studio — 로컬 PC에서 OpenAI 호환 API로 LLM·임베딩 모델을 서빙하는 도구. 노트북은 base_url만 로컬 서버로 바꿔 ChatOpenAI/OpenAIEmbeddings를 그대로 쓴다.

공통 인프라 — 노트북이 책과 갈라지는 첫 지점

원서는 코랩에서 OpenAIEmbeddings()(OpenAI API)와 ChatOpenAI(model="gpt-4o")(상용 API)를 쓴다. 이 강의의 노트북은 전부 로컬 LM Studio 서버로 임베딩과 LLM을 대체한다. base_url만 바꾸면 나머지 LangChain 코드는 그대로다.

from langchain_openai import OpenAIEmbeddings, ChatOpenAI

# LM Studio는 OpenAI 호환 엔드포인트를 로컬(:12345)에 연다. 키 검증을 안 하므로 api_key는 더미.
LMSTUDIO_BASE_URL = "http://host.docker.internal:12345/v1"

embeddings = OpenAIEmbeddings(
    model="embedding-8b:sl",
    base_url=LMSTUDIO_BASE_URL,
    api_key="lm-studio",
    # tiktoken으로 토큰을 세는 대신 원문 문자열을 그대로 전송한다. 로컬 모델은 OpenAI 토크나이저와
    # 어휘가 달라, 이 옵션 없이는 길이 계산이 어긋나 요청이 깨진다.
    check_embedding_ctx_length=False,
    # 로컬 서버는 한 번에 받는 텍스트가 많으면 연결이 끊긴다. 8개씩 나눠 보내 안정성을 확보.
    chunk_size=8,
)

llm = ChatOpenAI(
    model="qwen3.6-35b:mm",
    base_url=LMSTUDIO_BASE_URL,
    api_key="lm-studio",
    temperature=0,
    # Qwen3 계열은 답변 전에 <think> 추론 토큰을 크게 소모한다. 여기서 아끼면 추론만 하다
    # 답변이 빈 문자열로 잘린다 — 그래서 최소 2048, 긴 생성이 필요하면 4096으로 넉넉히 잡는다.
    max_tokens=2048,
)

로컬 추론 모델이 만드는 <think>...</think> 블록은 최종 출력에 섞이면 안 되므로, 모든 노트북이 같은 유틸로 걷어낸다.

import re

def strip_think(text: str) -> str:
    """추론 모델의 <think>...</think> 블록을 제거해 순수 답변만 남긴다."""
    if text is None:
        return ""
    return re.sub(r"(^|<think>).*?</think>", "", text, flags=re.DOTALL).strip()

책 vs 강의 — 왜 로컬인가: 책은 gpt-4o/OpenAI 임베딩으로 개념 전달에 집중한다. 강의는 API 비용·데이터 외부 유출 없이 온프레미스로 돌릴 수 있는 구성을 택했다. 대가로 (1) 추론 모델의 <think> 토큰을 걷어내야 하고 (2) max_tokens를 넉넉히 잡아야 하며 (3) 임베딩 배치를 작게 쪼개야 하는 로컬 특유의 함정이 생긴다. 이 세 가지는 상용 API에서는 겪지 않는 실전 이슈다.

1. 부모-자식 분할 — 부모를 RAM이 아니라 DB에 (practice4-1)

핵심 아이디어: "정밀하게 찾되, 넓게 반환한다." 작은 자식 청크로 임베딩 검색해 정확히 매칭하고, 그 자식이 속한 큰 부모 청크를 답변 컨텍스트로 돌려준다.

분할기 두 개를 부모(큰 것)·자식(작은 것)으로 나눈다.

from langchain_text_splitters import RecursiveCharacterTextSplitter

# 부모는 크게 — 검색에는 안 쓰이고, 답변에 넣을 넓은 맥락으로만 반환된다.
parent_splitter = RecursiveCharacterTextSplitter(chunk_size=1000)
# 자식은 작게 — 작을수록 임베딩이 특정 개념에 집중돼 매칭 정밀도가 오른다.
child_splitter = RecursiveCharacterTextSplitter(chunk_size=200)

책과 갈라지는 지점 — 부모 저장소의 영속성

원서는 자식 벡터는 Chroma에, 부모 원문은 InMemoryStore(RAM)에 담는다. 문제는 RAM은 프로세스가 죽으면 사라진다는 것 — 서버를 재시작하면 부모 문서를 모두 잃고 재인덱싱해야 한다. 강의 노트북은 부모도 Chroma 컬렉션에 key-value로 영속화한다. 이를 위해 LangChain의 BaseStore 인터페이스를 Chroma로 구현한다.

from typing import Iterator, List, Optional, Sequence, Tuple
from langchain_core.stores import BaseStore
from langchain_core.documents import Document

class ChromaParentStore(BaseStore[str, Document]):
    """부모 문서를 Chroma에 key-value로 저장하는 문서 저장소.
    유사도 검색은 하지 않고 id로만 get/set/delete 하므로, 임베딩 API를 호출하지 않고
    더미(0) 벡터를 채워 넣는다 — 부모 저장에 임베딩 비용을 쓰지 않기 위함이다."""

    def __init__(self, client, collection_name: str, dim: int = 8):
        self._collection = client.get_or_create_collection(collection_name)
        self._dim = dim

    def mget(self, keys: Sequence[str]) -> List[Optional[Document]]:
        if not keys:
            return []
        res = self._collection.get(ids=list(keys), include=["documents", "metadatas"])
        found = {i: (d, m) for i, d, m in zip(res["ids"], res["documents"], res["metadatas"])}
        # 요청한 key 순서를 그대로 유지하고, 없는 key 자리엔 None을 채운다 —
        # ParentDocumentRetriever가 자식→부모를 매핑할 때 이 순서 계약에 의존한다.
        return [
            Document(page_content=found[k][0], metadata=found[k][1] or {}) if k in found else None
            for k in keys
        ]

    def mset(self, key_value_pairs: Sequence[Tuple[str, Document]]) -> None:
        if not key_value_pairs:
            return
        ids = [k for k, _ in key_value_pairs]
        texts = [v.page_content for _, v in key_value_pairs]
        metas = [v.metadata or {} for _, v in key_value_pairs]
        # 부모는 검색 대상이 아니므로 진짜 임베딩 대신 0 벡터를 넣는다. 저장·조회만 되면 충분하다.
        dummy_embeddings = [[0.0] * self._dim for _ in ids]
        self._collection.upsert(ids=ids, documents=texts, metadatas=metas, embeddings=dummy_embeddings)

    def mdelete(self, keys: Sequence[str]) -> None:
        if keys:
            self._collection.delete(ids=list(keys))

    def yield_keys(self, *, prefix: Optional[str] = None) -> Iterator[str]:
        res = self._collection.get(include=[])
        for k in res["ids"]:
            if prefix is None or k.startswith(prefix):
                yield k

BaseStore가 요구하는 4개 메서드(mget/mset/mdelete/yield_keys)만 구현하면 ParentDocumentRetriever는 이 저장소를 InMemoryStore와 똑같이 취급한다. 인터페이스 계약을 지키는 한 저장 매체는 자유롭게 갈아끼울 수 있다는 것이 요점이다.

from langchain_chroma import Chroma

vectorstore = Chroma(client=chroma_client, collection_name=CHILD_COLLECTION, embedding_function=embeddings)
parent_store = ChromaParentStore(chroma_client, PARENT_COLLECTION)

# langchain 1.x에서 ParentDocumentRetriever의 위치가 langchain_classic으로 이동했다.
# 두 버전 어디서 돌려도 깨지지 않게 폴백 임포트로 감싼다.
try:
    from langchain_classic.retrievers import ParentDocumentRetriever
except ImportError:
    from langchain.retrievers import ParentDocumentRetriever

retriever = ParentDocumentRetriever(
    vectorstore=vectorstore,      # 자식 청크 임베딩 저장 → 유사도 검색
    docstore=parent_store,        # 부모 원문 저장 → id로 조회
    child_splitter=child_splitter,
    parent_splitter=parent_splitter,
)
retriever.add_documents(docs)   # 자동으로 부모/자식으로 쪼개 각 저장소에 나눠 넣는다

검색해 보면 반환은 부모(넓은 맥락), 벡터스토어 직접 검색은 자식(정밀 매칭)임을 확인할 수 있다.

query = "What are the types of investments?"
retrieved_docs = retriever.invoke(query)          # 부모 문서 (넓은 맥락)
sub_docs = vectorstore.similarity_search(query)   # 자식 문서 (정밀 키워드)

정리 — 책의 InMemoryStore는 개념 시연용이다. 프로세스 재시작·다중 프로세스 공유가 필요한 운영 환경에서는 부모 저장소도 영속 매체(여기선 Chroma, 5주차 이후엔 PostgreSQL)로 두는 것이 맞다. BaseStore 추상화 덕분에 리트리버 코드는 한 줄도 바꾸지 않고 저장소만 교체된다.

2. 질의 변형 (1) — MultiQuery (practice4-2)

핵심 아이디어: 하나의 질문은 표현이 하나뿐이라 검색을 놓칠 수 있다. LLM이 같은 질문을 여러 관점으로 바꿔 각각 검색하고, 결과를 합쳐 커버리지를 넓힌다.

import logging

# MultiQueryRetriever가 만든 변형 질의는 INFO 레벨 로그로만 보인다. 이 설정이 없으면
# 어떤 질의로 검색했는지 확인할 수 없어, 결과가 이상할 때 원인 추적이 막힌다.
logging.basicConfig()
logging.getLogger("langchain.retrievers.multi_query").setLevel(logging.INFO)

try:
    from langchain_classic.retrievers.multi_query import MultiQueryRetriever
except ImportError:
    from langchain.retrievers import MultiQueryRetriever

retriever = MultiQueryRetriever.from_llm(
    retriever=vectorstore.as_retriever(),  # 각 변형 질의를 실제로 검색할 기본 검색기
    llm=llm,                               # 질의를 여러 관점으로 변형할 로컬 LLM
)

question = "주식 투자를 처음 시작하려면 어떻게 해야 하나요?"
unique_docs = retriever.invoke(question)   # 내부에서 N개 질의 생성 → 각각 검색 → 중복 제거

책과 갈라지는 지점 — 로컬 추론 모델의 질의 생성

원서는 gpt-4o로 깔끔한 변형 질의 3개를 받는다. 강의는 로컬 추론 모델(qwen3.6-35b:mm)을 쓰므로, 질의 생성 응답에도 <think> 블록이 섞일 수 있고 max_tokens가 부족하면 변형 질의가 잘린다. 최종 답변 단계에서도 strip_think로 걷어낸 뒤 출력한다.

try:
    from langchain_classic.chains import RetrievalQA
except ImportError:
    from langchain.chains import RetrievalQA

qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",   # 검색 문서를 가공 없이 통째로 프롬프트에 넣는 가장 단순한 방식
    retriever=retriever,
    return_source_documents=True,
)
result = qa_chain.invoke({"query": question})
print("답변:", strip_think(result["result"]))  # 로컬 모델의 추론 흔적을 제거하고 답만 출력

언제 쓰나: 질문이 모호하거나 여러 측면을 담을 때. 단일 질의로 놓치는 관련 문서를 다관점 질의로 포착한다. 대가는 질의 개수만큼 늘어나는 LLM 호출·검색 비용이다.

3. 질의 변형 (2) — HyDE (practice4-3)

핵심 아이디어: 짧은 질문은 실제 문서와 어휘·문체가 다르다. LLM에게 질문의 가상 답변 문서를 먼저 쓰게 한 뒤, 그 가상 문서를 임베딩해 검색한다. 답변은 문서와 같은 전문 용어·문체를 쓰므로 임베딩 공간에서 실제 문서와 더 가깝다.

흐름: 질문 → 가상 답변 생성 → 가상 답변으로 벡터 검색 → 검색 문맥 + 원 질문 → 최종 답변. 각 단계를 독립 체인으로 만들고 RunnableLambda로 잇는다.

from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableLambda

# 1. 가상 문서 생성 체인 — 질문에 직접 답하는 가짜 문서를 chunk_size 언저리 길이로 만든다.
def create_virtual_doc_chain():
    system = "당신은 고도로 숙련된 AI입니다."
    user = "주어진 질문 '{query}'에 대해 직접적으로 답변하는 가상의 문서를 생성하세요. 크기는 {chunk_size} 글자 언저리."
    prompt = ChatPromptTemplate.from_messages([("system", system), ("human", user)])
    return prompt | llm | StrOutputParser()

# 2. 검색 체인 — 질문이 아니라 '가상 문서'로 유사도 검색한다는 것이 HyDE의 전부다.
# RunnableLambda는 일반 파이썬 함수를 체인처럼 invoke로 부를 수 있게 감싸 주는 부품이다.
def create_retrieval_chain():
    return RunnableLambda(lambda x: retriever.invoke(x["virtual_doc"]))

def format_docs(docs):
    return "\n\n".join(doc.page_content for doc in docs)

# 3. 최종 응답 체인 — 검색으로 얻은 진짜 문서(context)와 원 질문으로 답한다.
# 가상 문서는 검색용 미끼일 뿐, 답변 근거로는 절대 쓰지 않는다 (환각 방지의 핵심).
def create_final_response_chain():
    final_prompt = ChatPromptTemplate.from_template(
        "다음 정보와 질문을 바탕으로 답변해주세요.\n컨텍스트: {context}\n질문: {question}\n답변:"
    )
    return final_prompt | llm

세 체인을 단계별로 잇는다.

def create_pipeline_with_logging():
    virtual_doc_chain = create_virtual_doc_chain()
    retrieval_chain = create_retrieval_chain()
    final_response_chain = create_final_response_chain()

    def virtual_doc_step(x):
        result = {"virtual_doc": virtual_doc_chain.invoke({"query": x["question"], "chunk_size": 200})}
        return {**x, **result}

    def retrieval_step(x):
        return {**x, "retrieved_docs": retrieval_chain.invoke(x)}

    def context_formatting_step(x):
        return {**x, "context": format_docs(x["retrieved_docs"])}

    def final_response_step(x):
        return final_response_chain.invoke(x)

    # 딕셔너리를 단계마다 키를 덧붙이며 흘려보낸다. 각 단계의 출력 키가 다음 단계의 입력 키와
    # 맞아야 파이프가 이어진다 — 이 키 계약이 어긋나는 것이 파이프라인 디버깅의 주된 원인이다.
    return (
        RunnableLambda(virtual_doc_step)
        | RunnableLambda(retrieval_step)
        | RunnableLambda(context_formatting_step)
        | RunnableLambda(final_response_step)
    )

pipeline = create_pipeline_with_logging()
response = pipeline.invoke({"question": "주식 시장의 변동성이 높을 때 투자 전략은 무엇인가요?"})
print(strip_think(response.content))

책과 갈라지는 지점 — 추론 모델의 max_tokens

가상 문서 생성은 200자 언저리의 제법 긴 출력을 요구한다. 로컬 추론 모델은 답변 전에 <think>로만 2000토큰 넘게 소모할 수 있어, max_tokens=2048이면 가상 문서가 빈 문자열로 잘린다. 노트북은 이 노트북에 한해 max_tokens=4096으로 올린다. gpt-4o에서는 겪지 않는, 로컬 추론 모델 특유의 실전 함정이다.

MultiQuery vs HyDE — 둘 다 질의 변형이지만 방향이 다르다. MultiQuery는 질문을 여러 개로 늘려 커버리지를 넓히고, HyDE는 질문을 답변 형태로 바꿔 질문-문서 표현 간극을 좁힌다. 전문 용어가 많아 질문과 문서 어휘가 크게 다른 도메인엔 HyDE가, 다각도 검색이 필요한 모호한 질문엔 MultiQuery가 유리하다.

4. 희소 검색 — pgvector로 직접 만드는 BM25 (practice4-4)

여기서부터가 이 강의가 원서를 가장 크게 넘어서는 대목이다.

책과 갈라지는 지점 — 인메모리 BM25의 확장성 한계

원서는 BM25Retriever.from_documents(docs, preprocess_func=kiwi_tokenize)로 프로세스 메모리 안에 BM25 인덱스를 만든다. 코퍼스 전체를 RAM에 올리고 매 질의마다 전수 스캔하므로, 억 단위 레코드에는 애초에 못 쓴다. 강의는 같은 BM25를 PostgreSQL 인프라 레벨에서 다시 만든다. 핵심은 "질의 시점에 원문을 다시 스캔하지 않고, 미리 저장한 통계만 조합해 점수를 낸다"는 것이다.

저장해야 할 세 가지 통계

BM25 점수를 매 질의마다 재계산하지 않으려면 세 값을 미리 저장·유지해야 한다.

필요한 값	저장 위치
TF — 문서 안에서 토큰이 몇 번 등장했나	순색인(forward index): `(문서, 토큰) → 횟수`
DF — 토큰이 몇 개 문서에 등장했나 → IDF	역색인(inverted index): `토큰 → 문서 수`
문서 길이, 코퍼스 평균 문서 길이(avgdl)	문서 테이블 `doc_len` + 단일 행 통계 테이블

문서 테이블은 tsvector(형태소 분석 결과)와 GIN 인덱스를 갖는다. korean 설정은 pgvector 이미지에 포함된 mecab-ko/textsearch_ko 확장이 제공한다 — 원서의 Kiwi 형태소 분석에 대응하는, DB 내장 한국어 토크나이저다.

DDL_DOCS_TABLE = """
CREATE EXTENSION IF NOT EXISTS textsearch_ko;

CREATE TABLE IF NOT EXISTS practice4_bm25_docs (
    id       SERIAL PRIMARY KEY,
    content  TEXT NOT NULL,
    page     INTEGER,
    tsv      TSVECTOR,      -- mecab-ko 기반 korean 설정으로 형태소 분석한 결과
    doc_len  INTEGER        -- 이 문서의 총 토큰 수 (BM25 길이 정규화용)
);
CREATE INDEX IF NOT EXISTS idx_practice4_bm25_docs_tsv
    ON practice4_bm25_docs USING GIN (tsv);   -- 토큰 포함 문서를 빠르게 후보로 좁히는 역색인
"""

트리거로 색인을 증분 유지

문서 추가·수정·삭제 때마다 세 통계를 트리거로 자동 갱신한다. 색인 유지 비용이 "바뀐 문서 1건" 규모로 끝나고, 코퍼스 전체를 다시 훑을 필요가 없다.

DDL_TRIGGERS = """
-- BEFORE INSERT/UPDATE: 원문 → tsvector 형태소 분석, 총 토큰 수(doc_len) 계산
CREATE OR REPLACE FUNCTION practice4_bm25_docs_before() RETURNS TRIGGER AS $$
BEGIN
    NEW.tsv := to_tsvector('korean', NEW.content);
    SELECT COALESCE(SUM(cardinality(positions)), 0)
      INTO NEW.doc_len
      FROM unnest(NEW.tsv) AS u(lexeme, positions, weights);
    RETURN NEW;
END;
$$ LANGUAGE plpgsql;

-- AFTER INSERT: 새 문서의 토큰을 순색인에 추가, 역색인 DF를 +1, 코퍼스 통계 갱신
CREATE OR REPLACE FUNCTION practice4_bm25_docs_after_insert() RETURNS TRIGGER AS $$
BEGIN
    INSERT INTO practice4_bm25_term_freq (doc_id, token, tf)
    SELECT NEW.id, lexeme, cardinality(positions)
      FROM unnest(NEW.tsv) AS u(lexeme, positions, weights)
    ON CONFLICT (doc_id, token) DO UPDATE SET tf = EXCLUDED.tf;

    INSERT INTO practice4_bm25_term_df (token, df)
    SELECT lexeme, 1
      FROM unnest(NEW.tsv) AS u(lexeme, positions, weights)
    ON CONFLICT (token) DO UPDATE SET df = practice4_bm25_term_df.df + 1;

    UPDATE practice4_bm25_stats
       SET doc_count = doc_count + 1, total_len = total_len + NEW.doc_len
     WHERE id = 1;
    RETURN NEW;
END;
$$ LANGUAGE plpgsql;
"""
# DELETE 트리거는 위의 역연산(순색인 행 삭제, DF -1, df<=0 토큰 정리, 통계 감소)이고,
# UPDATE 트리거는 OLD를 지운 뒤 NEW를 넣는 DELETE+INSERT 조합이다.

노트북이 짚는 미묘한 함정: DF를 내리는 UPDATE와 0 이하 토큰을 지우는 DELETE를 한 문장의 데이터 변경 CTE로 합치면 안 된다. PostgreSQL은 같은 문장 안의 변경 CTE들을 같은 스냅숏에서 평가하므로, 방금 UPDATE로 내린 행을 뒤이은 DELETE가 못 찾는다. 그래서 두 문장으로 분리한다. 재적재는 TRUNCATE 대신 DELETE를 쓴다 — TRUNCATE는 행 단위 트리거를 실행하지 않아 파생 통계가 원복되지 않기 때문이다.

저장된 통계만으로 BM25 점수 계산

Okapi BM25 공식은 다음과 같다.

score(d,q) = Σ  IDF(t) · [ f(t,d)·(k1+1) ] / [ f(t,d) + k1·(1 - b + b·|d|/avgdl) ]
             t∈q

IDF(t) = ln( (N - n(t) + 0.5) / (n(t) + 0.5) + 1 )

f(t,d)=TF(순색인), n(t)=DF(역색인), N·avgdl(통계 테이블), k1(기본 1.2)·b(기본 0.75)는 TF 포화·길이 정규화 강도. 미리 채운 세 테이블만 조인하면 원문 재스캔이 없다.

DDL_BM25_FUNC = """
CREATE OR REPLACE FUNCTION practice4_bm25_search(
    q text, k integer DEFAULT 5, bm25_k1 float8 DEFAULT 1.2, bm25_b float8 DEFAULT 0.75
) RETURNS TABLE(doc_id integer, content text, page integer, score float8) AS $$
    WITH stats AS (
        SELECT doc_count::float8 AS n,
               CASE WHEN doc_count > 0 THEN total_len::float8 / doc_count ELSE 0 END AS avgdl
          FROM practice4_bm25_stats WHERE id = 1
    ),
    qtokens AS (   -- 질의도 같은 korean 설정으로 토큰화해야 문서 토큰과 매칭된다
        SELECT DISTINCT lexeme AS token
          FROM unnest(to_tsvector('korean', q)) AS u(lexeme, positions, weights)
    ),
    qidf AS (      -- 질의 토큰별 IDF. 코퍼스에 없는 토큰은 df=0으로 처리(LEFT JOIN)
        SELECT qt.token,
               ln( (s.n - COALESCE(d.df,0) + 0.5) / (COALESCE(d.df,0) + 0.5) + 1 ) AS idf
          FROM qtokens qt CROSS JOIN stats s
          LEFT JOIN practice4_bm25_term_df d ON d.token = qt.token
    ),
    scored AS (    -- 문서별로 질의 토큰 점수를 합산 = BM25 점수
        SELECT tf.doc_id,
               SUM( qidf.idf * (tf.tf * (bm25_k1 + 1))
                    / (tf.tf + bm25_k1 * (1 - bm25_b + bm25_b * doc.doc_len / NULLIF(s.avgdl,0))) ) AS score
          FROM practice4_bm25_term_freq tf
          JOIN qidf ON qidf.token = tf.token
          JOIN practice4_bm25_docs doc ON doc.id = tf.doc_id
          CROSS JOIN stats s
         GROUP BY tf.doc_id
    )
    SELECT scored.doc_id, doc.content, doc.page, scored.score
      FROM scored JOIN practice4_bm25_docs doc ON doc.id = scored.doc_id
     ORDER BY scored.score DESC LIMIT k;
$$ LANGUAGE sql STABLE;
"""

DB 내장 ts_rank_cd(cover density)를 기준선으로 먼저 비교하는데, 이건 plainto_tsquery가 질의 토큰을 모두 포함(AND)하는 문서만 찾아 결과가 0건이 되기도 한다. BM25는 토큰이 일부만 겹쳐도 점수를 누적(OR에 가까움)하고 TF 포화·길이 정규화까지 반영해 더 관대하고 정교하다.

마지막으로 BaseRetriever로 감싸 LangChain 파이프라인에 연결한다. 랭킹 연산은 전부 DB에서 끝나고 파이썬은 상위 k건만 받으므로, 코퍼스를 메모리에 올리지 않는다.

from langchain_core.retrievers import BaseRetriever
from langchain_core.callbacks import CallbackManagerForRetrieverRun

class PgSqlFunctionRetriever(BaseRetriever):
    dsn: str
    sql_func: str
    k: int = 4

    def _get_relevant_documents(self, query, *, run_manager):
        with psycopg.connect(self.dsn) as conn:
            with conn.cursor() as cur:
                # 질의는 항상 파라미터 바인딩(%s)으로 넘긴다 — 함수명은 코드가 지정한 상수뿐이라
                # SQL 인젝션 표면을 사용자 입력 쪽에서 완전히 차단한다.
                cur.execute(f"SELECT doc_id, content, page, score FROM {self.sql_func}(%s, %s)",
                            (query, self.k))
                rows = cur.fetchall()
        return [Document(page_content=c,
                         metadata={"doc_id": i, "page": p, "score": float(s)})
                for i, c, p, s in rows]

bm25_retriever = PgSqlFunctionRetriever(dsn=PG_DSN, sql_func="practice4_bm25_search", k=2)

관찰 — 키워드 검색만의 한계: 코퍼스에는 "발행주식총수(A) 보통주 13,602,977"(489쪽) 같은 정답 문단이 있다. 하지만 질의 "총 발행량"은 원문의 표현("발행주식의 총수")과 겹치는 단어가 적어, BM25 상위권에서도 정답 문단이 k=2에 잘려 나간다. 질의와 문서의 표현이 다르면 순수 키워드 검색은 정답을 놓친다 — 다음 절의 밀집 검색이 이 문제를 푼다.

5. 밀집 검색 — pgvector HNSW + MRL 차원 축소 (practice4-5)

책과 갈라지는 지점 — FAISS 파일 vs pgvector 서버

원서는 FAISS.from_documents()로 인메모리 인덱스를 만들고 save_local/load_local로 파일에 저장한다. FAISS는 결국 한 프로세스 메모리에 얹히므로, 재시작하거나 다른 프로세스에서 쓰려면 파일 전체를 다시 읽어야 하고, 억 단위로 커지면 한 장비 메모리에 안 올라간다. 강의는 pgvector에 진짜 ANN 인덱스인 HNSW를 구축해 DB에 영구 저장한다. 어떤 프로세스에서 접속하든 같은 인덱스를 그대로 쓴다.

참고로 원서 4장(3.2절)이 짚는 FAISS의 load_local(..., allow_dangerous_deserialization=True) 위험 — 인덱스가 pickle이라 여는 순간 심어진 코드가 실행될 수 있어 남의 파일엔 절대 켜면 안 되는 옵션 — 은 pgvector 방식에서는 아예 존재하지 않는다. 벡터가 DB 행으로 저장되기 때문이다.

차원 상한 문제와 MRL truncation

pgvector의 HNSW/IVFFlat 인덱스는 vector 타입 기준 2000차원(절반 정밀도 halfvec이면 4000차원)이 상한이다. 그런데 로컬 임베딩 모델 embedding-8b:sl(Qwen3 임베딩 계열)은 4096차원을 내놓는다. 노트북은 먼저 4096차원 그대로 HNSW를 만들어 실제로 오류가 나는 것을 확인시킨 뒤, 해결책을 제시한다.

이 모델은 MRL(Matryoshka Representation Learning)로 학습돼, 벡터의 앞쪽 차원만 잘라내도 그 자체로 유효한 저차원 임베딩이 된다. 학습된 투영 행렬이나 랜덤 프로젝션 없이 그냥 앞에서 자르면(truncation) 된다. 코사인 유사도는 방향만 보므로 길이가 줄어도 재정규화가 필요 없다(pgvector 코사인 연산자가 내부 정규화).

from langchain_core.embeddings import Embeddings

PROJ_DIM = 1024  # HNSW(vector) 상한 2000보다 넉넉히 작게

class TruncatedEmbeddings(Embeddings):
    """MRL로 학습된 임베딩의 앞쪽 out_dim개 차원만 사용해 pgvector 인덱스 상한에 맞추는 래퍼.
    MRL이 아닌 일반 임베딩을 이렇게 자르면 유사도가 크게 훼손되므로, 이 방식은 MRL 모델 전용이다."""

    def __init__(self, base: Embeddings, out_dim: int):
        self.base = base
        self.out_dim = out_dim

    def embed_documents(self, texts):
        return [v[: self.out_dim] for v in self.base.embed_documents(texts)]

    def embed_query(self, text):
        return self.base.embed_query(text)[: self.out_dim]

embeddings = TruncatedEmbeddings(base_embeddings, out_dim=PROJ_DIM)  # 4096 → 앞쪽 1024

langchain_postgres v2로 HNSW 구축

from langchain_postgres import PGEngine, PGVectorStore
from langchain_postgres.v2.indexes import HNSWIndex, DistanceStrategy

engine = PGEngine.from_connection_string(url=PG_ASYNC_URL)

# overwrite_existing=True: 매 실행마다 DROP 후 CREATE → 몇 번을 다시 돌려도 중복 적재가 없다.
engine.init_vectorstore_table(table_name=DENSE_TABLE, vector_size=PROJ_DIM, overwrite_existing=True)

vectorstore = PGVectorStore.create_sync(
    engine=engine,
    embedding_service=embeddings,
    table_name=DENSE_TABLE,
    distance_strategy=DistanceStrategy.COSINE_DISTANCE,
    k=2,
)
vectorstore.add_documents(docs)

# HNSW 파라미터: m=이웃 연결 수, ef_construction=인덱스 구축 시 탐색 폭.
# 값이 클수록 재현율↑·구축 비용↑ — 정확도와 속도의 트레이드오프 손잡이다.
vectorstore.apply_vector_index(
    HNSWIndex(name="practice4_dense_docs_hnsw",
              distance_strategy=DistanceStrategy.COSINE_DISTANCE, m=16, ef_construction=64)
)
dense_retriever = vectorstore.as_retriever(search_kwargs={"k": 2})

관찰 — 의미 검색이 표현 차이를 넘는다: 4절의 BM25(k=2)가 놓친 정답을, 이 밀집 검색은 1위로 328쪽 문단("Ⅳ. 발행주식의 총수 ... 13,602,977")을 찾아내 RetrievalQA가 "13,602,977주"를 정확히 답했다. 질의에 "발행량"이 없어도 "발행주식의 총수"를 의미 유사도로 찾은 것이다. 다만 밀집 검색은 표현이 완전히 다른 숫자·고유명사·법조문 번호의 정확 매칭엔 오히려 약하다 — 그래서 다음 절의 앙상블이 필요하다.

6. 앙상블 검색 — BM25 + 밀집을 RRF로 융합 (practice4-6)

희소는 정확 키워드에 강하고 표현이 다르면 놓치며, 밀집은 의미엔 강하고 정확 매칭엔 약하다. 두 약점이 상보적이므로 함께 쓴다. EnsembleRetriever는 점수 체계가 다른 두 결과를 RRF(Reciprocal Rank Fusion) — 각 문서의 순위 역수 합 — 로 융합한다. 점수 절댓값이 아니라 순위만 쓰므로, BM25 점수와 코사인 거리처럼 스케일이 전혀 다른 신호도 안정적으로 합쳐진다.

bm25_retriever = PgSqlFunctionRetriever(dsn=PG_DSN, sql_func="practice4_bm25_search", k=2)
dense_retriever = vectorstore.as_retriever(search_kwargs={"k": 2})

try:
    from langchain_classic.retrievers import EnsembleRetriever
except ImportError:
    from langchain.retrievers import EnsembleRetriever

# weights: 도메인이 키워드 중심이면 BM25를, 의미 중심이면 밀집을 높인다. 책과 동일하게 0.5:0.5.
ensemble_retriever = EnsembleRetriever(retrievers=[bm25_retriever, dense_retriever], weights=[0.5, 0.5])

리트리버 인터페이스가 동일하므로, 강의의 pgvector 리트리버든 책의 FAISS 리트리버든 EnsembleRetriever에 똑같이 꽂힌다. 실제 저장·색인·랭킹만 PostgreSQL로 옮겼을 뿐 LangChain 조립 코드는 책과 같다.

관찰 — 앙상블의 재현율 한계: BM25가 놓친 정답을 밀집 검색이 1위로 찾았고, RRF가 그 문단을 앙상블 결과에 살려 답변에 반영했다. "둘 중 하나만 맞아도 산다"가 앙상블의 힘이다. 하지만 분명한 한계가 있다 — EnsembleRetriever는 각 리트리버가 이미 잘라낸 상위 k건만 합친다. 두 리트리버 모두의 상위 k에서 빠진 문서는 앙상블도 복구하지 못한다. 즉 앙상블 재현율은 "개별 리트리버 중 최선"을 넘지 못한다. 실무 해법은 각 리트리버의 k를 최종 반환 수보다 넉넉히 키워 합친 뒤 재순위화(rerank)하는 것 — 다음 절의 주제다.

7. 리랭킹 — 로컬 Qwen3-Reranker로 2단계 검색 완성 (practice4-7)

책과 갈라지는 지점 — gpt-4o LLM 리랭킹 vs 로컬 리랭커

원서는 두 가지 리랭킹을 보여 준다. (1) gpt-4o에 "1~10점으로 관련성을 매겨라"를 프롬프트해 점수를 받는 LLM 리랭킹 — 정확하지만 문서마다 API 호출이라 느리고 비싸다. (2) sentence-transformers의 CrossEncoder("cross-encoder/ms-marco-MiniLM-L-12-v2") — 로컬 경량 크로스 인코더.

강의는 로컬 크로스 인코더 노선을 택하되 최신 모델 Qwen3-Reranker-0.6B를 쓴다. 여기서 노트북이 짚는 실전 함정이 중요하다: 이 모델을 CrossEncoder(...)로 로딩하면 분류 헤드(score.weight)가 체크포인트에 없어 무작위 초기화되고 점수가 무의미해진다. 그래서 모델 카드의 원조 방식인 AutoModelForCausalLM 기반 yes/no 스코어링을 쓴다 — 질의-문서 쌍을 프롬프트로 넣고, 모델이 다음 토큰으로 yes/no를 낼 로그확률을 관련성 점수로 삼는다.

from transformers import AutoModelForCausalLM, AutoTokenizer

class Qwen3Reranker:
    """질의-문서 쌍에 대해 'yes'/'no' 다음 토큰 확률을 관련성 점수로 사용하는 리랭커."""

    def __init__(self, model_name="Qwen/Qwen3-Reranker-0.6B", device=DEVICE):
        self.tokenizer = AutoTokenizer.from_pretrained(model_name, padding_side="left")
        self.model = AutoModelForCausalLM.from_pretrained(model_name, dtype=torch.float16).to(device).eval()
        # 'yes'/'no'의 토큰 id를 미리 잡아 둔다 — score()에서 이 두 위치의 logit만 비교한다.
        self.token_false_id = self.tokenizer.convert_tokens_to_ids("no")
        self.token_true_id = self.tokenizer.convert_tokens_to_ids("yes")
        self.max_length = 8192
        # 모델 카드가 지정한 고정 프롬프트 틀. 마지막에 빈 <think></think>를 붙여 추론을 건너뛰고
        # 곧바로 yes/no를 생성하게 유도한다 (추론 토큰 낭비 없이 판정만 받기 위함).
        self.prefix = ('<|im_start|>system\nJudge whether the Document meets the requirements based on '
                       'the Query and the Instruct provided. Note that the answer can only be "yes" or "no".'
                       '<|im_end|>\n<|im_start|>user\n')
        self.suffix = "<|im_end|>\n<|im_start|>assistant\n<think>\n\n</think>\n\n"
        self.prefix_tokens = self.tokenizer.encode(self.prefix, add_special_tokens=False)
        self.suffix_tokens = self.tokenizer.encode(self.suffix, add_special_tokens=False)
        self.task = "주어진 검색 질의에 대해 답변이 될 수 있는 관련 문서를 찾아라"

    def _format(self, query, doc):
        return f"<Instruct>: {self.task}\n<Query>: {query}\n<Document>: {doc}"

    @torch.no_grad()
    def score(self, query, doc_texts):
        pairs = [self._format(query, d) for d in doc_texts]
        inputs = self._process(pairs)                       # 고정 prefix/suffix로 감싸 패딩
        batch_scores = self.model(**inputs).logits[:, -1, :]  # 마지막 토큰 위치의 logit만 사용
        true_vector = batch_scores[:, self.token_true_id]
        false_vector = batch_scores[:, self.token_false_id]
        # yes/no 두 logit만 뽑아 log_softmax → yes 확률을 관련성 점수로. 두 후보만 비교하므로
        # 어휘 전체 softmax보다 안정적이다.
        stacked = torch.stack([false_vector, true_vector], dim=1)
        probs = torch.nn.functional.log_softmax(stacked, dim=1)
        return probs[:, 1].exp().tolist()

2단계 검색 파이프라인

앞 절들과 달리 각 리트리버의 후보 폭을 k=4로 넓혀 앙상블한 뒤(최대 8건), 리랭커로 top_n=2를 정밀 선별한다. 리랭커에 넘기기 전 중복 청크를 제거한다.

bm25_retriever = PgSqlFunctionRetriever(dsn=PG_DSN, sql_func="practice4_bm25_search", k=4)
dense_retriever = vectorstore.as_retriever(search_kwargs={"k": 4})
ensemble_retriever = EnsembleRetriever(retrievers=[bm25_retriever, dense_retriever], weights=[0.5, 0.5])

reranker = Qwen3Reranker()

def reranking_documents(query, docs_, top_n=2):
    scores = reranker.score(query, [d.page_content for d in docs_])
    scored_docs = sorted(zip(docs_, scores), key=lambda x: x[1], reverse=True)
    return [d for d, _ in scored_docs[:top_n]]

이 2단계 검색기를 다시 BaseRetriever로 감싸 RetrievalQA에 그대로 연결한다.

from typing import Any, List
from pydantic import ConfigDict, Field

class RerankingRetriever(BaseRetriever):
    """1차 후보 검색기(앙상블)의 결과를 Qwen3 리랭커로 재순위화해 top_n건만 반환."""

    # pydantic은 검증할 줄 아는 타입만 필드로 받는다. 리랭커·리트리버 같은 외부 객체를 담으려면
    # 이 허용 설정이 필요하다.
    model_config = ConfigDict(arbitrary_types_allowed=True)

    base_retriever: Any = Field(description="1차 후보 검색기 (앙상블)")
    reranker_top_n: int = 2

    def _get_relevant_documents(self, query, *, run_manager):
        candidates_ = self.base_retriever.invoke(query)
        seen_, unique_ = set(), []
        for d in candidates_:
            if d.page_content not in seen_:   # 두 리트리버가 같은 청크를 뽑을 수 있어 중복 제거
                seen_.add(d.page_content)
                unique_.append(d)
        return reranking_documents(query, unique_, top_n=self.reranker_top_n)

reranking_retriever = RerankingRetriever(base_retriever=ensemble_retriever, reranker_top_n=2)

관찰 — 리랭킹이 앙상블의 "합치기만 하는" 한계를 보완한다: 앙상블(각 k=2)은 상위 2건씩을 순위 기반으로 합칠 뿐 합친 결과를 다시 평가하지 않았다. 이 노트북은 후보 폭을 k=4로 넓혀 더 많이 모은 뒤(최대 8건) 리랭커가 질의와 각 후보를 실제로 다시 비교해 top_n=2를 골랐다. 결과는 동일하게 "13,602,977주"였지만, 이번엔 "각 리트리버 상위 k가 우연히 정답을 포함했는가"에 의존하지 않고 더 넓은 후보에서 관련성 기준으로 선별했다는 점이 다르다. 두 리트리버 상위 k에 정답이 없을 수 있는 실무 상황에서 특히 유효하다.

보안 메모

practice4-7은 허깅페이스 토큰을 .hf_token 파일에서 읽어 "하드코딩 금지"를 지킨다. DB 접속 정보 같은 credential도 같은 원칙 — 소스에 직접 박지 말고 파일·환경변수로 분리한다. (학습자료에는 실제 비밀번호를 옮기지 않으므로 위 예제의 PG_DSN은 접속 정보를 별도 로드하는 것으로 가정하라.)

세 노트북(4-4·4-5·4-6)이 책과 다른 점 — 요약표

축	책 (인메모리)	강의 (pgvector 서버)
희소 검색	`BM25Retriever`(Kiwi), 프로세스 메모리, 매 질의 전수 스캔	트리거로 증분 유지되는 순색인/역색인/통계 + SQL로 계산하는 BM25
밀집 검색	FAISS 인메모리, `save_local`/`load_local` 파일	pgvector HNSW, DB에 영구 저장, pickle 위험 없음
차원 처리	OpenAI 임베딩(고정 차원)	4096차원 → MRL truncation으로 1024차원(인덱스 상한 회피)
앙상블	`EnsembleRetriever(bm25, faiss)`	`EnsembleRetriever(bm25, dense)` — 인터페이스 동일
리랭킹	gpt-4o LLM 리랭킹(API 과금) 또는 CrossEncoder	로컬 Qwen3-Reranker-0.6B(yes/no 로그확률)
확장성	검색기마다 코퍼스 전체를 RAM에 적재	색인·데이터가 DB에 있어 다중 프로세스 공유, 증분 갱신·ANN으로 흡수
LLM/임베딩	OpenAI 상용 API	LM Studio 로컬(`<think>`·max_tokens·배치 처리 필요)

핵심: BaseRetriever/EnsembleRetriever 같은 LangChain 인터페이스는 그대로 두고, 실제 저장·색인·랭킹·추론만 프로덕션 인프라로 옮긴 것이 이 강의의 정체성이다.

실무 체크리스트

[ ] 긴 문서에서 정밀 검색 + 넓은 맥락이 필요하면 부모-자식 분할. 운영 환경이면 부모 저장소도 영속 매체로.
[ ] 질문이 모호하거나 다각도면 MultiQuery, 질문-문서 어휘 간극이 크면 HyDE.
[ ] 로컬 추론 모델을 쓰면 <think> 제거 + max_tokens 넉넉히 + 임베딩 배치 작게.
[ ] 억 단위 코퍼스면 인메모리 BM25/FAISS 대신 DB 인프라(트리거 증분 색인 + ANN 인덱스).
[ ] 임베딩 차원이 pgvector 상한(2000/halfvec 4000)을 넘으면, MRL 모델이면 truncation·아니면 랜덤 프로젝션.
[ ] 앙상블만으로 재현율이 부족하면 후보 k를 키워 합친 뒤 리랭커로 정밀 선별(2단계 검색).
[ ] 로컬 리랭커는 로딩 방식 확인 — 분류 헤드가 없는 모델은 CrossEncoder 로딩이 무의미할 수 있다.
[ ] credential·토큰은 소스에 하드코딩 금지, 파일·환경변수로 분리.